Introduction
Welcome!
Code Examples
Day 1
Challenge: KISS
Solution: KISS
Day 2
Challenge: Type Annotations
Solution: Type Annotations
Day 3
Challenge: Decoupling
Solution: Decoupling
Day 4
Challenge: DRY
Solution: DRY
Day 5
Challenge: String Formatting
Solution: String Formatting
Day 6
Challenge: Law Of Demeter
Solution: Law Of Demeter
Day 7
Challenge: Better Discounts
Solution: Better Discounts
Day 8
Challenge: Payment Strategy
Solution: Payment Strategy
Day 9
Challenge: Plugins
Solution: Plugins
Day 10
Challenge: Object Oriented To Functional
Solution: Object Oriented To Functional
Day 11
Challenge: Cohesion
Solution: Cohesion
Day 12
Challenge: MVP
Solution: MVP
Day 13
Challenge: Inheritance
Solution: Inheritance
Day 14
Challenge: Abstraction
Solution: Abstraction
Day 15
Challenge: Higher-Order Functions
Solution: Higher-Order Functions
Day 16
Challenge: Configuration
Solution: Configuration
Day 17
Challenge: Concurrency
Solution: Concurrency
Day 18
Challenge: Refactoring
Solution: Refactoring
Day 19
Challenge: Itertools
Solution: Itertools
Day 20
Challenge: Inappropriate Intimacy
Solution: Inappropriate Intimacy
Wrap Up
End of Part 1

Hi Andreas, hope you're doing well!
I came up with this solution, which is more similar to Arjan's v1 version.
************ Start code**************
from dataclasses import dataclass
from typing import Any
from time import perf_counter
import requests
import asyncio
API_KEY = "123456"
@dataclass
class UrlTemplateClient:
template: str
def get(self, data: dict[str, Any]) -> Any:
url = self.template.format(**data)
response = requests.get(url, timeout=5)
response.raise_for_status() # Raise an exception if the request failed
return response.json()
async def a_get(self, data: dict[str, Any]) -> Any:
return await asyncio.to_thread(self.get, data)
class CityNotFoundError(Exception):
pass
async def get_capital(country: str) -> str:
client = UrlTemplateClient(template="https://restcountries.com/v3/name/{country}")
response = await client.a_get({"country": country})
# The API can return multiple matches, so we just return the capital of the first match
return response[0]["capital"][0]
async def get_forecast(city: str) -> dict[str, Any]:
client = UrlTemplateClient(
template=f"http://api.openweathermap.org/data/2.5/weather?q={city}&appid={API_KEY}"
)
response = await client.a_get({"city": city})
if "main" not in response:
raise CityNotFoundError(
f"Couldn't find weather data. Check '{city}' if it exists and is correctly spelled.\n"
)
return response
def get_temperature(full_weather_forecast: dict[str, Any]) -> float:
temperature = full_weather_forecast["main"]["temp"]
return temperature - 273.15 # convert from Kelvin to Celsius
async def async_main() -> None:
countries = ["United States of America", "Australia", "Japan", "France", "Brazil","Argentina"]
start = perf_counter()
country_capitals = await asyncio.gather(*[get_capital(country) for country in countries])
capital_wheater_forecasts = await asyncio.gather(*[get_forecast(city) for city in country_capitals])
for index, country in enumerate(countries):
print(f"The capital of {country} is {country_capitals[index]}")
print(
f"The current temperature in {country_capitals[index]} is {get_temperature(capital_wheater_forecasts[index]):.1f} °C."
)
print(f'Total elapsed time: {perf_counter() - start}')
if __name__ == "__main__":
asyncio.run(async_main())
************ End code**************
I've really enjoyed this challenge.
I still have to get used to using zip.
Have a nice week end!
Hi! I'm doing well, hope you are well!
Nice solution! I think this is a nice improvement! I have some smaller remarks to make this even better:
* For
a_getmethod, try using more detailed naming and not shorthands. Usually, it is not beneficial to add a label that says that the function is async, it does not add anything to what the method does, only how it is executed.*
get_capitalwill not always return a string because it does an API call, and it might not return a country. I would suggest that you assert or check with a guard clause if something is returned. If not, either raise an exception or None* For the
getmethod in the dataclass, this one can have type annotationdict[str, Any]since the .json()methods returnsdict[str, Any]Excellent recommendations! I've made the following changes in the portion of code you mentioned in your remarks:
@dataclass
class UrlTemplateClient:
template: str
def get_http(self, data: dict[str, Any]) -> dict[str, Any]:
url = self.template.format(**data)
response = requests.get(url, timeout=5)
response.raise_for_status() # Raise an exception if the request failed
return response.json()
async def get(self, data: dict[str, Any]) -> Any:
return await asyncio.to_thread(self.get_http, data)
class RestCountryApiError(Exception):
pass
class CityNotFoundError(Exception):
pass
async def get_capital(country: str) -> str:
client = UrlTemplateClient(template="https://restcountries.com/v3/name/{country}")
try:
response = await client.get({"country": country})
if response and isinstance(response, list) and "capital" in response[0]:
# The API can return multiple matches, so we just return the capital of the first match
return response[0]["capital"][0]
else:
raise RestCountryApiError(f"Invalid missing response format or missing 'capital' data")
except Exception as error:
raise RestCountryApiError(f"There was a error calling restcountries.com API: {error}")
Thanks Andreas!
Great improvement, Alberto!
Hi, I just want to say that when usage of concurrency is limited and simple, a nice solution is to use the ThreadPoolExecuter or ProcessPoolExecutor objects for the concurrent.futures module (standard library). The API is very nice and it can be used as a contect manager.
I think that if you need to do the same operation in parallel (do some image processing on a collection of images, do many API calls in parallel etc...), than this is a good solution. Also, no need for async code all the way, which avoids some complexification of the code.
Asyncio is great when the program is more complex and the concurrent operations are all over the place and different
Interesting! I have never used ThreadPoolExecuter or ProcessPoolExecutor. I will do a smaller project to try it out!
Asyncio also is widely known, which is always good (It is only a plus and not an argument in my opinion :) )
For me, it is still hard to know whether or not I will be able to implement a concurrent solution in my own projects (I have one that would terriby benefit from this)...but still don't get the whole picture.
But this exercise has put a little bit more light on my understanding of this topic. This is something I definitely need to work on harder.
Thanks Arjan! Great challenge.
Hi Juan José, concurrent programming is quite a hard topic to wrap your mind around. The best way I found to learn is by practicing with it a lot. In your own project, you could experiment with switching to concurrent programming by making the methods/functions asynchronous. You can then experiment with asyncio.gather to group a few of these concurrent functions and see what the effect is on performance. It's a step-by-step process :).
Thanks for the tip! I am currently trying to switch it to concurrent. It is a small utility that analyzes some backtests. I have a function with a for loop. Each pass does all the analysis. I think I need to change the design to one function that does the analysis for just one backtest and make it asynchronous. I suspect I will have to use asyncio.gather to group all the analyzing together...
Don't know if it will work...but for sure I will get more insight into this topic.
I feel like this challenge single handedly took my understanding to the next level! I also love the great comments people post and your awesome replies Arjan. This whole 30 day challenge is Brilliant
Similar answer here. But I have kept the UrlTemplateClient code much similar to the "before" code, and I have created a decorator to make the function async.
def asynchronizer(func: Callable[..., Any]) -> Callable[..., Any]:
····async def inner(*args: Any, **kwargs: Any):
········return await asyncio.to_thread(func, *args, **kwargs)
····return inner
@asynchronizer
def get(self, data: dict[str, Any]) -> Any:
····url = self.template.format(**data)
····response = requests.get(url, timeout=5)
····response.raise_for_status() # Raise an exception if the request failed
····return response.json()
That's a great use for a decorator. Thanks for sharing this!